/*
Copyright 2024.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"math/rand"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"

	v1 "k8s.io/api/batch/v1"
	core "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/remotecommand"
	"k8s.io/client-go/util/retry"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	chemicalv1 "chemical.ai/rabbitjob/api/v1"
)

// RabbitJobReconciler reconciles a RabbitJob object
type RabbitJobReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

var (
	NeedDelete operate = "delete"
	NeedCreate operate = "create" //livness metric
	jobs       map[string]int
)

// +kubebuilder:rbac:groups=chemical.chemical.kubebuilder.io,resources=rabbitjobs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=chemical.chemical.kubebuilder.io,resources=rabbitjobs/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=chemical.chemical.kubebuilder.io,resources=rabbitjobs/finalizers,verbs=update

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
// the RabbitJob object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.18.4/pkg/reconcile
func (r *RabbitJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	rabbitJob := chemicalv1.RabbitJob{}
	// TODO(user): your logic here
	if err := r.Client.Get(ctx, req.NamespacedName, &rabbitJob); err != nil {
		log.Log.Error(err, fmt.Sprintf("fail to get Job from %s", req.Namespace))
	}
	//create Jobs
	jobs = make(map[string]int)
	log.Log.Info("success to get rabbitJob", "replicas' number", rabbitJob.Spec.Replicas)
	replica := 0
	for _ = range int(rabbitJob.Spec.Replicas) {
		log.Log.Info("creating one job")
		err := createJob(r, ctx, &rabbitJob, jobs)
		handleErr(err, fmt.Sprintf("fail to create Job %s in cluster", rabbitJob.ObjectMeta.Name))

		replica++
	}
	//add created number to field status
	rabbitJob.Status.Replicas = int32(replica)
	secret, err := linkToSecret("chemairs")
	handleErr(err, "fail to get secret")
	srtOption := rabbitOption{
		name:     "hello", //queue's name
		username: base64.StdEncoding.EncodeToString(secret.Data["username"]),
		passwd:   base64.StdEncoding.EncodeToString(secret.Data["password"]),
		standard: int(rabbitJob.Spec.Lenthreshold),
		ctx:      ctx,
		missing:  float32(rabbitJob.Spec.Miss) / 100,
		url:      "rabbitmq-service", //service-name
	}
	montor := make(chan operate)
	wg := sync.WaitGroup{}
	wg.Add(1)
	//monitor thread, and also producer
	go func() {
		for {

			if operate, ok := judgeIfNeedScale(srtOption); ok {
				//need to do some operate
				montor <- operate
				time.Sleep(20 * time.Second)
				rabbitJob.Status.LastGettingTime = &metav1.Time{
					Time: time.Now(),
				}
			}
		}
	}()
	//consumer thread
	go func() {
		{
			{
				operate := <-montor
				if operate == NeedCreate {
					createJob(r, ctx, &rabbitJob, jobs)
				} else if operate == NeedDelete {
					job := dealWhichJob(rabbitJob, jobs)
					sendSigterm(*job, jobs)
				}
			}
		}
	}()
	return ctrl.Result{}, nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *RabbitJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
	if err := mgr.GetFieldIndexer().IndexField(context.Background(), &v1.Job{}, ".metadata.controller", func(rawObj client.Object) []string {
		job, ok := rawObj.(*v1.Job)
		if !ok {
			fmt.Print("the object you recieve is not be Job")
		}
		owner := metav1.GetControllerOf(job)
		if owner == nil {
			return nil
		}
		if owner.APIVersion != chemicalv1.GroupVersion.String() || owner.Kind != "RabbitJob" {
			return nil
		}
		return []string{owner.Name}
	}); err != nil {
		return err
	}
	return ctrl.NewControllerManagedBy(mgr).
		For(&chemicalv1.RabbitJob{}).
		Owns(&v1.Job{}).
		Complete(r)
}

func createJob(r *RabbitJobReconciler, ctx context.Context, rabbitJob *chemicalv1.RabbitJob, jobs map[string]int) error {
	job := buildJob(*rabbitJob)
	if err := r.Client.Create(ctx, job); err != nil {
		return err
	}
	rabbitJob.Status.JobNames = append(rabbitJob.Status.JobNames, job.Name)
	jobs[job.ObjectMeta.Name] = 50
	return nil
}

func buildJob(rabbitjob chemicalv1.RabbitJob) *v1.Job {
	jobname := rabbitjob.Name + "-" + randomStrBylength(5)
	job := v1.Job{
		ObjectMeta: metav1.ObjectMeta{
			Name:      jobname,
			Namespace: rabbitjob.Namespace,
		},
		Spec: v1.JobSpec{
			Completions:    getint32(1),
			ManagedBy:      &rabbitjob.Name,
			CompletionMode: (*v1.CompletionMode)(getpointofstr("Indexed")),
			Template: core.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{"controller": jobname},
				},
				Spec: core.PodSpec{
					Containers: []core.Container{
						{
							Image:           rabbitjob.Spec.Job.Template.Spec.Containers[0].Image,
							Name:            rabbitjob.Spec.Job.Template.Spec.Containers[0].Name,
							ImagePullPolicy: core.PullIfNotPresent,
						},
					},
					RestartPolicy: core.RestartPolicyOnFailure,
				},
			},
			BackoffLimit:            rabbitjob.Spec.Job.BackoffLimit,
			TTLSecondsAfterFinished: getint32(20), //20 second after pod become complete, delete pod , and implement elegance delete
		},
	}
	return &job
}

// deal with which job need to be deleted
func dealWhichJob(rabbitJob chemicalv1.RabbitJob, jobs map[string]int) *v1.Job {
	//init the map
	max := 0
	jobname := ""
	for job, _ := range jobs {
		value := Acticity(joboption{
			Namespace: rabbitJob.Namespace,
			Name:      job,
		})
		jobs[job] = value
		if max < value {
			max = value
			jobname = job
		}
	}
	if jobname == "" {
		return &v1.Job{}
	}
	job, err := linkToKube().BatchV1().Jobs(rabbitJob.Namespace).Get(context.TODO(), jobname, metav1.GetOptions{})
	if err != nil {
		handleErr(err, "fail to get job")
	}

	return job
}

type joboption struct {
	Namespace string
	Name      string
}

func Acticity(job joboption) int {
	clientset := linkToKube()
	pods, err := clientset.CoreV1().Pods(job.Namespace).List(context.TODO(), metav1.ListOptions{
		LabelSelector: "job-name=" + job.Name,
	})
	handleErr(err, "failt get pod list from rabbitJob")
	pod := pods.Items[0]
	options := podoption{
		Podname:       pod.Name,
		containername: pod.Spec.Containers[0].Name,
		Namespace:     job.Namespace,
	}
	log, err := getPodLog(clientset, options)
	handleErr(err, "fail to get log from container")
	//2024-10-10T08:29:04.595+08:00  INFO 6 --- [    Thread-8510] c.c.a.r.a.s.i.TaskManagementServiceImpl  : search task complete of 3a1585fd-8cb7-66cb-2035-7a60cf133c77 done!
	slices := strings.Split(log, "\n")
	for i := len(slices) - 1; i >= 0; i-- {
		if strings.Contains(slices[i], "task complete") {
			//已经被关闭
			t1 := judgeTime(slices[i])
			t2 := time.Now()
			dura := t2.Sub(t1)
			return int(dura.Seconds())
		} else if strings.Contains(slices[i], "start task") {
			//正在执行任务
			return 0
		}
	}
	return 0
}

// get pod's log by given namespace
// 方法需要测试，不清楚获取到的日志格式，目前按照常规，有换行，格式正常的情况做假设
func getPodLog(client *kubernetes.Clientset, pod podoption) (string, error) {
	opts := &core.PodLogOptions{
		Container: pod.containername,
		Follow:    true,
		TailLines: getint64(1000), //处理一千条消息
	}
	req := client.CoreV1().Pods(pod.Namespace).GetLogs(pod.Podname, opts)
	logReader, err := req.Stream(context.TODO())
	if err != nil {
		return "", err
	}
	defer logReader.Close()
	buf := new(bytes.Buffer)
	_, err = buf.ReadFrom(logReader)
	if err != nil {
		return "", err
	}
	return buf.String(), nil
}

func linkToSecret(secretname string) (*core.Secret, error) {
	clientset := linkToKube()
	//just RabbitMq server's namespace
	secret, err := clientset.CoreV1().Secrets("default").Get(context.TODO(), secretname, metav1.GetOptions{})
	if err != nil {
		return nil, fmt.Errorf("fail to get secret from apiserver, probably by service account %w", err)
	}
	return secret, nil
}

type rabbitOption struct {
	name     string
	url      string
	username string
	passwd   string
	ctx      context.Context
	standard int
	missing  float32
}

type podoption struct {
	Podname       string
	Namespace     string
	containername string
}

// send sigterm go special container and kill the thread
func sendSigterm(job v1.Job, jobs map[string]int) {
	podoptions := podoption{
		Namespace: job.Namespace,
		Podname:   job.Name,
	}
	clientset := linkToKube()
	pod, err := clientset.CoreV1().Pods(podoptions.Namespace).Get(context.TODO(), podoptions.Podname, metav1.GetOptions{})
	handleErr(err, "fail to get pod resource")
	podoptions.containername = pod.Spec.Containers[0].Name
	cmd := []string{"/bin/sh", "-c", "kill -s SIGTERM 1"} // kill the PID = 1
	req := clientset.CoreV1().RESTClient().Post().Resource("pod").Name(podoptions.Podname).Namespace(podoptions.Namespace).SubResource("exec")
	config := &rest.Config{}
	execopt := &core.PodExecOptions{
		Container: podoptions.containername,
		Command:   cmd,
		Stdin:     false,
		Stdout:    true,
		Stderr:    true,
		TTY:       false,
	}
	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
		execreq := req.VersionedParams(execopt, metav1.ParameterCodec)
		exec, err := remotecommand.NewSPDYExecutor(config, "GET", execreq.URL())
		if err != nil {
			return fmt.Errorf("fail to create a new exec request(sigterm) %w", err)
		}
		reader, writer, err := os.Pipe()
		if err != nil {
			return fmt.Errorf("fail to create pip %w", err)
		}
		go func() {
			io.Copy(os.Stdout, reader)
		}()
		err = exec.Stream(remotecommand.StreamOptions{
			Stdin:             nil,
			Stdout:            writer,
			Stderr:            writer,
			Tty:               false,
			TerminalSizeQueue: nil,
		})
		writer.Close()
		if err != nil {
			return fmt.Errorf("fail to execting command %w", err)
		}
		return nil
	})
	delete(jobs, job.ObjectMeta.Name)
	if err != nil {
		handleErr(err, fmt.Sprintf("fail to send sigterm to container %s", podoptions.containername))
	}
}

type QueueInfo struct {
	Messages       int `json:"messages"`
	MessgeReady    int `json:"message_ready"`
	MessageUnacked int `json:"message_unacked"`
}

type operate string

// http://localhost:15672/api/queues/%2F/
func judgeIfNeedScale(option rabbitOption) (operate, bool) {
	url := "http://" + GetIpFromURL(option) + ":15672/api/queues/" + option.name
	req, err := http.NewRequestWithContext(option.ctx, "GET", url, nil)
	handleErr(err, "fail to initial http request")
	req.SetBasicAuth(option.username, option.passwd)
	client := &http.Client{}
	resp, err := client.Do(req)
	handleErr(err, "fail to connect RabbitMq server")
	defer resp.Body.Close()
	var queueInfo QueueInfo
	err = json.NewDecoder(resp.Body).Decode(&queueInfo)
	handleErr(err, "fail to decoder the responese of rabbitMq server")
	if AbsInt(queueInfo.MessgeReady-option.standard) <= (int(option.missing) * option.standard) {
		//the missing is below to given
		return "", false
	} else {
		if queueInfo.MessgeReady > option.standard {
			return NeedCreate, true
		}
		return NeedDelete, true
	}
}

func GetIpFromURL(option rabbitOption) string {
	client := linkToKube()
	service, err := client.CoreV1().Services("default").Get(context.TODO(), option.url, metav1.GetOptions{})
	handleErr(err, "fail to get ip from servicename")
	return service.Spec.ClusterIP
}

func judgeTime(str string) time.Time {
	strs := strings.Split(str, "  INFO ")
	layout := "2006-01-02T15:04:05.999Z07:00"
	t, err := time.Parse(layout, strs[0])
	handleErr(err, "fail to parse string to Object time")
	return t
}

func linkToKube() *kubernetes.Clientset {
	config, err := clientcmd.BuildConfigFromFlags("", os.Getenv("KUBECONFIG"))
	handleErr(err, "profile kubeconfig not found ,fail to link apiserver")
	clientset, err := kubernetes.NewForConfig(config)
	handleErr(err, "fail to init clientset")
	return clientset
}
func AbsInt(n int) int {
	if n > 0 {
		return n
	}
	return -n
}

func getint32(num int) *int32 {
	i := int32(num)
	return &i
}

func getint64(num int) *int64 {
	i := int64(num)
	return &i
}

// make random string by given length
func randomStrBylength(length int) string {
	example := "qwipoesdpalfqsnfckeajsbnfhpuerq"
	rand.Seed(time.Now().UnixNano())
	b := make([]byte, length)
	for i := range b {
		b[i] = example[rand.Intn(len(example))]
	}
	return string(b)
}

func getpointofstr(str string) *string {
	return &str
}
func handleErr(err error, str string) {
	if err != nil {
		log.Log.Info(err.Error(), str)
		os.Exit(1)
	}
}
